/*
 *  Copyright 2009 Lucas Nazário dos Santos
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */
package net.sourceforge.retriever.executor;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import net.sourceforge.retriever.analyzer.Analyzer;
import net.sourceforge.retriever.feedback.Feedback;
import net.sourceforge.retriever.fetcher.dns.DNSResolver;
import net.sourceforge.retriever.filter.Filter;
import net.sourceforge.retriever.frontier.Frontier;

/**
 * TODO Write javadoc.
 */
public class ThreadExecutor implements Executor {


	private final int corePoolSize;

	private final int maximumPoolSize;

	private final long keepAliveTime;

	private final TimeUnit timeUnit;

	private final int queueSize;

	private final ThreadPoolExecutor threadPool;

	/**
	 * Specify a pool of threads.
	 * 
	 * @param corePoolSize The number of threads inside the pool's core.
	 * @param maximumPoolSize The maximum number of threads the pool will 
	 *        handle at once.
	 * @param keepAliveTime For how long time a thread will live after it
	 *        ceases responding.
	 * @param timeUnit The time unit for the keepAliveTime property.
	 * @param queueSize The amount of threads that can be enqueued waiting for
	 * 		  a chance to actually get into the pool.
	 */
	public ThreadExecutor(final int corePoolSize, final int maximumPoolSize, final long keepAliveTime, final TimeUnit timeUnit, final int queueSize) {
		this.corePoolSize = corePoolSize;
		this.maximumPoolSize = maximumPoolSize;
		this.keepAliveTime = keepAliveTime;
		this.timeUnit = timeUnit;
		this.queueSize = queueSize;

		final ArrayBlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(queueSize);
		final BlockThenRunPolicy blockingPolicy = new BlockThenRunPolicy(60, TimeUnit.SECONDS);

		this.threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, timeUnit, queue, blockingPolicy);
	}

	public void execute(final Frontier urlFrontier, final Filter filter, final Analyzer analyzer, final Feedback feedback, final Object emptyFrontierLock, final DNSResolver dnsResolver) {
		this.threadPool.execute(new URLParser(urlFrontier, filter, analyzer, feedback, emptyFrontierLock, dnsResolver));
	}

	public boolean isExecuting() {
		return this.threadPool.getActiveCount() > 0;
	}

	public void shutdown() {
		this.threadPool.shutdown();
	}

	public ThreadExecutor create() {
		return new ThreadExecutor(this.corePoolSize, this.maximumPoolSize, this.keepAliveTime, this.timeUnit, this.queueSize);
	}

	private class BlockThenRunPolicy implements RejectedExecutionHandler {

	    private final long maxBlockTime;

	    private final TimeUnit blockTimeoutUnit;

	    private BlockThenRunPolicy(final long maxBlockTime, final TimeUnit blockTimeoutUnit) {
	    	this.maxBlockTime = maxBlockTime;
	    	this.blockTimeoutUnit = blockTimeoutUnit;
	    }

	    public void rejectedExecution(final Runnable task, final ThreadPoolExecutor executor) {           

	    	final BlockingQueue<Runnable> queue = executor.getQueue();
	        boolean taskSent = false;
	        int attempts = 0;

	        while (!taskSent && attempts++  < 3) {
	            if (executor.isShutdown()) {
	                throw new RejectedExecutionException("ThreadPoolExecutor has shutdown while attempting to offer a new task.");
	            }

	            try {
	                if (queue.offer(task, maxBlockTime, blockTimeoutUnit)) {
	                    taskSent = true;
	                }
	            } catch (final InterruptedException e) {
	            }
	        }

	        if (attempts == 3) {
	        	// TODO Log
	        }
	    }
	}
}